package yunjiao.javatutorials.apache.kafka.component;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import yunjiao.javatutorials.apache.kafka.model.InventoryUpdateEvent;

/**
 * 库存服务消费者
 *
 * @author yangyunjiao
 */
@Service
@Slf4j
public class InventoryEventConsumer {

    @KafkaListener(
            topics = "${app.kafka.topics.inventory-update}",
            groupId = "${app.kafka.consumer-groups.inventory-service}",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void consumeInventoryUpdateEvent(InventoryUpdateEvent event,
                                            Acknowledgment ack) {
        try {
            log.info(">>> Processing inventory update: productId={}, operation={}, quantity={}",
                    event.getProductId(), event.getOperation(), event.getQuantity());

            // 模拟库存更新逻辑
            updateInventory(event);

            ack.acknowledge();
            log.info(">>> Inventory update processed: productId={}", event.getProductId());

        } catch (Exception e) {
            log.error("Error processing inventory update: productId={}",
                    event.getProductId(), e);
            // 库存操作失败，可能需要回滚或告警
        }
    }

    private void updateInventory(InventoryUpdateEvent event) {
        // 实际的库存更新逻辑
        // 这里可以调用库存服务或直接操作数据库
        if (InventoryUpdateEvent.OperationType.DECREASE.equals(event.getOperation())) {
            // 扣减库存
            log.info(">>> Decreasing inventory for product: {}, quantity: {}",
                    event.getProductId(), event.getQuantity());
        } else if (InventoryUpdateEvent.OperationType.INCREASE.equals(event.getOperation())) {
            // 增加库存（用于取消订单等情况）
            log.info("Increasing inventory for product: {}, quantity: {}",
                    event.getProductId(), event.getQuantity());
        }

        // 模拟处理时间
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
